
/***************************************************************************
 *   fecab 
 *                                                                         *
 *   Copyright (C) 2010 by kuerant                                         *
 *                         <kuerant@gmail.com>                             *
 *                                                                         *
 *   This library is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Library General Public License as       *
 *   published by the Free Software Foundation; either version 2 of the    *
 *   License, or (at your option) any later version.                       *
 *                                                                         *
 *   This library is distributed in the hope that it will be usefull, but  *
 *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
 *   Library General Public License for more details.                      *
 *                                                                         *
 *   You should have received a copy of the GNU Library General Public     *
 *   License along with this library; if not, write to the Free Software   *
 *   Foundation, Inc., 59 Temple Place - Suite 330, Boston,                *
 *   MA 02111-1307, USA                                                    *
 ***************************************************************************/

#include <cstdio>
#include <cstdlib>
#include <sstream>

#include <boost/filesystem.hpp>

using boost::system::error_code;
using boost::system::system_category;
using boost::system::system_error;

#include "bocat/common/ini.hpp"
#include "bocat/common/logging.hpp"

#include "fecab/store.hpp"

#include "sleepystore.hpp"


namespace	fecab	{

#define	N_CHECKPOINT_INTERVAL	3000000

///////////////////////////////////////////////////////////////////////
//  class SleepyStore
//

SleepyStore::SleepyStore( const char* config_filename )
			:	config( config_filename ), 
				dbp_cache( config.ncache_tables)
				
{

	// check directory
	boost::filesystem::path	envhome_path( config.envhome );
	if ( !boost::filesystem::exists( envhome_path ) ) {
		//char	msg[512];
		//snprintf(msg, sizeof(msg) - 1, "Env home '%s' does NOT exist", config.envhome.c_str());
		//throw std::runtime_error( msg );
		if ( !boost::filesystem::create_directories( envhome_path ) ) {
            std::ostringstream  msg;
            msg << "fail to create env-home directory '" << config.envhome << "'";
            throw boost::filesystem::filesystem_error( msg.str(), envhome_path, error_code(errno, system_category) );
		}
	}
	if ( !boost::filesystem::is_directory( envhome_path ) ) {
		char	msg[512];
		snprintf(msg, sizeof(msg) - 1, "Env home '%s' is NOT a directory", config.envhome.c_str());
		throw std::runtime_error( msg );
	}

    boost::filesystem::path dbconfig_path = envhome_path / "DB_CONFIG";
    if ( !boost::filesystem::exists( dbconfig_path ) ) {
        // create default DB_CONFIG
        createDefaultDbConfig( dbconfig_path.string() );
    }

	dbenv_p = NULL;
	if ( config.in_recovery ) {
		bocat::common::Logger*  logger_p = bocat::common::getLogger();

		logger_p->info(__LINE__, __FILE__, "recovering ...");
		recoverDbEnv();
		logger_p->info(__LINE__, __FILE__, "recover done.");
	}
	openDbEnv();

	_count_put_or_erase = 0;

	//_dbt_data_buf_p = NULL;
	//_dbt_data_buf_size = 0;

	_dbt_data_buf_size = 1024;
	_dbt_data_buf_p = new char[ _dbt_data_buf_size ];
}

SleepyStore::~SleepyStore() 
{
	// close all tables
	closeAllTables();
	purgeDbp();

	if (_count_put_or_erase > 0) {
		dbenv_p->txn_checkpoint( 0, 0, 0 );
		_count_put_or_erase = 0;
	}

	// close db environment
	closeDbEnv();

	if (NULL != _dbt_data_buf_p) {
		delete [] _dbt_data_buf_p;
		_dbt_data_buf_p = NULL;
		_dbt_data_buf_size = 0;
	}
} 


void
SleepyStore::createDefaultDbConfig( const std::string filename )
{
    std::ofstream   ofs;

    try {
        ofs.open( filename.c_str() );

        ofs << std::endl
            << std::endl << "set_cachesize 0 524288000 0"
            << std::endl << ""
            << std::endl << "set_lg_regionmax  8388608"
            << std::endl << "set_lg_max        10485760"
            << std::endl << "set_lg_bsize      2097152"
            << std::endl << "#set_lg_dir        /path/to/var/confab/dblog"
            << std::endl << ""
            << std::endl << "set_lk_max_objects  65536"
            << std::endl << "set_lk_max_locks    131072"
            << std::endl << "set_lk_max_lockers  131072"
            << std::endl << ""
            << std::endl << "set_flags         DB_TXN_WRITE_NOSYNC "
            << std::endl;

        ofs.close();
    }
    catch( std::exception& e) {
        if (ofs.is_open()) {
            ofs.close();
        }

        std::ostringstream  msg;
        msg << "fail to create default config " << filename;
        throw std::runtime_error( msg.str() );
    }
}


///////////////////////////////////////////////////////////////////////
//


int     
SleepyStore::put( std::string table,
				  const char* key,    u_int32_t key_size,
				  const char* data,   u_int32_t data_size )
{
	dbp_type	dbp = this->openTable( table );

	Dbt key_dbt, data_dbt;
	key_dbt.set_data( const_cast< char * >(key) );
	key_dbt.set_size( key_size );
	data_dbt.set_data( const_cast< char * >(data) );
	data_dbt.set_size( data_size );

	DbTxn *txn = NULL;
	dbenv_p->txn_begin(NULL, &txn, 0);
	try {
		dbp->put(txn, &key_dbt, &data_dbt, 0);
		txn->commit(0);

		++_count_put_or_erase;
		if (_count_put_or_erase > N_CHECKPOINT_INTERVAL) {
			dbenv_p->txn_checkpoint( 0, 0, 0 );
			_count_put_or_erase = 0;
		}
	} catch (DbException &e) {
		txn->abort();

		char	msg[ 1024 ];
		snprintf(msg, sizeof(msg) - 1, "fail to insert data into table '%s' (ENV:%s DBFILE:%s) -- %s", table.c_str(), config.envhome.c_str(), config.dbfile.c_str(), e.what() );
		throw std::runtime_error( msg );
	}

	return	0;
}

int     
SleepyStore::put( const char* table,
				  const char* key,    u_int32_t key_size,
				  const char* data,   u_int32_t data_size )
{
	return	put( std::string(table), key, key_size, data, data_size );
}

int     
SleepyStore::put( u_int32_t   pin,
				  const char* key,    u_int32_t key_size,
				  const char* data,   u_int32_t data_size )
{
	std::string	table = getTableName( pin );

	return	put( table, key, key_size, data, data_size );
}


int     
SleepyStore::get( std::string table,
				  const char* key,    u_int32_t key_size,
				  const char* buf,    u_int32_t buf_size )
{
	dbp_type	dbp = this->openTable( table );

	Dbt key_dbt, data_dbt;
	key_dbt.set_data( const_cast< char * >(key) );
	key_dbt.set_size( key_size );

	data_dbt.set_data( const_cast< char * >(buf) );
	data_dbt.set_ulen( buf_size );
	data_dbt.set_flags(DB_DBT_USERMEM);

	try {
		int	rc = dbp->get(NULL, &key_dbt, &data_dbt, 0);
		if (rc != 0) {
			if (DB_NOTFOUND == rc) {
				return 0;	// NOT FOUND
			}
			else {
			}
		}
	} catch (DbException &e) {
		if ( DB_NOTFOUND == e.get_errno() ) {
			return 0;	// NOT FOUND
		}

		char	msg[ 1024 ];
		snprintf(msg, sizeof(msg) - 1, "fail to get data from table '%s' (ENV:%s DBFILE:%s) -- %s", table.c_str(), config.envhome.c_str(), config.dbfile.c_str(), e.what() );
		throw std::runtime_error( msg );
	}

	return	data_dbt.get_size();
}

int     
SleepyStore::get( const char* table,
				  const char* key,    u_int32_t key_size,
				  const char* buf,    u_int32_t buf_size )
{
	return	get( std::string(table), key, key_size, buf, buf_size );
}


int     
SleepyStore::get( u_int32_t   pin,
				  const char* key,    u_int32_t key_size,
				  const char* buf,    u_int32_t buf_size )
{
	std::string	table = getTableName( pin );

	return	get( table, key, key_size, buf, buf_size );
}


int     
SleepyStore::erase( std::string table,
				  const char* key,    u_int32_t key_size ) 
{
	dbp_type	dbp = this->openTable( table );

	Dbt key_dbt;
	key_dbt.set_data( const_cast< char * >(key) );
	key_dbt.set_size( key_size );

	DbTxn *txn = NULL;
	dbenv_p->txn_begin(NULL, &txn, 0);
	try {
		dbp->del(txn, &key_dbt, 0);
		txn->commit(0);

		++_count_put_or_erase;
		if (_count_put_or_erase > N_CHECKPOINT_INTERVAL) {
			dbenv_p->txn_checkpoint( 0, 0, 0 );
			_count_put_or_erase = 0;
		}
	} catch (DbException &e) {
		txn->abort();

		char	msg[ 1024 ];
		snprintf(msg, sizeof(msg) - 1, "fail to delete data from table '%s' (ENV:%s DBFILE:%s) -- %s", table.c_str(), config.envhome.c_str(), config.dbfile.c_str(), e.what() );
		throw std::runtime_error( msg );
	}

	return	0;
}


int     
SleepyStore::erase( const char* table,
				    const char* key,    u_int32_t key_size )
{
	return	erase( std::string(table), key, key_size );
}

int     
SleepyStore::erase( u_int32_t   pin,
				    const char* key,    u_int32_t key_size )
{
	std::string	table = getTableName( pin );

	return	erase( table, key, key_size );
}


std::string     
SleepyStore::getTableName( u_int32_t pin )
{
	char	buf[512];

	u_int32_t	table_id = getTableId( pin );
	snprintf( buf, sizeof(buf) - 1, "X%04X", table_id );

	return std::string( buf );
}

u_int32_t
SleepyStore::getTableId( u_int32_t pin )
{
	return pin % config.ntables;
}

///////////////////////////////////////////////////////////////////////
//

int     
SleepyStore::put( const std::string& table,
				  const std::string& key,    
				  const std::string& data )
{
	return	put( table, key.c_str(), key.size(), data.c_str(), data.size() );
}

int     
SleepyStore::put( u_int32_t   pin,
				  const std::string& key,
				  const std::string& data )
{
	std::string	table = getTableName( pin );

	return	put( table, key, data );
}

int     
SleepyStore::get( const std::string& table,
				  const std::string& key,
					    std::string& buf )
{
	boost::mutex::scoped_lock	_scoped_lock( _dbt_data_buf_mutex );

	dbp_type	dbp = this->openTable( table );

	Dbt key_dbt, data_dbt;
	memset( &key_dbt, 0, sizeof(Dbt) );
	key_dbt.set_data( const_cast< char * >(key.c_str()) );
	key_dbt.set_size( key.size() );

	memset( &data_dbt, 0, sizeof(Dbt) );
	data_dbt.set_data( _dbt_data_buf_p );
	data_dbt.set_ulen( _dbt_data_buf_size );
	data_dbt.set_flags(DB_DBT_USERMEM);

	try {
		int	rc = dbp->get(NULL, &key_dbt, &data_dbt, 0);
		if (rc != 0) {
			if (DB_NOTFOUND == rc) {
				return 0;	// NOT FOUND
			}
			else {
			}
		}
	} catch (DbException &e) {
		if ( DB_NOTFOUND == e.get_errno() ) {
			return 0;	// NOT FOUND
		}

		char	msg[ 1024 ];
		snprintf(msg, sizeof(msg) - 1, "fail to get data from table '%s' (ENV:%s DBFILE:%s) -- %s", table.c_str(), config.envhome.c_str(), config.dbfile.c_str(), e.what() );
		throw std::runtime_error( msg );
	}

	buf.assign( static_cast< char * >(data_dbt.get_data()), data_dbt.get_size() );

	return	buf.size();
}

int     
SleepyStore::get( u_int32_t   pin,
				  const std::string& key,
				 	    std::string& buf )
{
	std::string	table = getTableName( pin );

	return	get( table, key, buf );
}

int     
SleepyStore::erase( const std::string& table,
					const std::string& key )
{
	return	erase( table, key.c_str(), key.size() );
}

int     
SleepyStore::erase( u_int32_t   pin,
				    const std::string& key )
{
	std::string	table = getTableName( pin );

	return	erase( table, key );
}


///////////////////////////////////////////////////////////////////////
//

void
SleepyStore::openDbEnv(void)
{
	if (NULL == dbenv_p) {
		dbenv_p = new DbEnv(0);

		u_int32_t	env_flags = DB_CREATE     | // If the environment does not exist, create it.
								DB_INIT_LOCK  | // Initialize locking
								DB_INIT_LOG   | // Initialize logging
								DB_INIT_MPOOL | // Initialize the cache
								DB_INIT_TXN   | // Initialize transactions
								DB_PRIVATE    |
								DB_READ_UNCOMMITTED |
								DB_THREAD;

		try {
			dbenv_p->open( config.envhome.c_str(), env_flags, 0664 );

			// Turn off system buffering of Berkeley DB log files to avoid double caching.
			////dbenv_p->log_set_config( DB_LOG_DIRECT, 1 );

			// Berkeley DB will automatically remove log files that are no longer needed.
			dbenv_p->log_set_config( DB_LOG_AUTO_REMOVE, 1 );

			dbenv_p->log_set_config( DB_LOG_ZERO, 1 );
		}
		catch(DbException& e) {
			delete dbenv_p;
			dbenv_p = NULL;

			char	msg[ 1024 ];
			snprintf(msg, sizeof(msg) - 1, "fail to open DbEnv '%s' -- %s", config.envhome.c_str(), e.what() );
			throw std::runtime_error( msg );
		}
	}
}


void
SleepyStore::closeDbEnv(void)
{
	if (NULL != dbenv_p) {
		try {
			dbenv_p->close( 0 );
		}
		catch(DbException& e) {
			char	msg[ 1024 ];
			snprintf(msg, sizeof(msg) - 1, "fail to close DbEnv '%s' : %s", config.envhome.c_str(), e.what() );
			throw std::runtime_error( msg );
		}

		delete dbenv_p;
		dbenv_p = NULL;
	}
}


void
SleepyStore::recoverDbEnv(void)
{
	if (NULL == dbenv_p) {
		dbenv_p = new DbEnv(0);

		u_int32_t	env_flags = DB_CREATE     | // If the environment does not exist, create it.
								DB_INIT_LOCK  | // Initialize locking
								DB_INIT_LOG   | // Initialize logging
								DB_INIT_MPOOL | // Initialize the cache
								DB_INIT_TXN   | // Initialize transactions
								DB_PRIVATE    |
								//DB_FAILCHK    |
								//DB_REGISTER   |
								//DB_RECOVER;
								DB_RECOVER_FATAL;

		try {
			dbenv_p->open( config.envhome.c_str(), env_flags, 0664 );

			// Turn off system buffering of Berkeley DB log files to avoid double caching.
			////dbenv_p->log_set_config( DB_LOG_DIRECT, 1 );

			// Berkeley DB will automatically remove log files that are no longer needed.
			dbenv_p->log_set_config( DB_LOG_AUTO_REMOVE, 1 );

			dbenv_p->log_set_config( DB_LOG_ZERO, 1 );
		}
		catch(DbException& e) {
			delete dbenv_p;
			dbenv_p = NULL;

			char	msg[ 1024 ];
			snprintf(msg, sizeof(msg) - 1, "fail to open DbEnv '%s' for recovery -- %s", config.envhome.c_str(), e.what() );
			throw std::runtime_error( msg );
		}

		try {
			dbenv_p->close( 0 );
		}
		catch(DbException& e) {
			char	msg[ 1024 ];
			snprintf(msg, sizeof(msg) - 1, "fail to close DbEnv '%s' for recovery -- %s", config.envhome.c_str(), e.what() );
			throw std::runtime_error( msg );
		}

		delete dbenv_p;
		dbenv_p = NULL;
	}
}


///////////////////////////////////////////////////////////////////////
//


SleepyStore::dbp_type
SleepyStore::openTable( const std::string& table_name )
{
	boost::mutex::scoped_lock	_scoped_lock( this->dbp_cache_mutex );

	dbp_type	dbp;

	if ( dbp_cache.exists( table_name ) ) {
		dbp = dbp_cache.get( table_name );
		return dbp;
	}

	dbp = this->getDbp();
//fprintf(stderr, "%s:%d %s() dbp=%p\n", __FILE__, __LINE__, __func__, dbp);
	try {
		u_int32_t db_flags = DB_CREATE | DB_AUTO_COMMIT;

		dbp->set_bt_minkey( config.btree_minkey );

//fprintf(stderr, "%s:%d %s() dbp=%p open  table %s\n", __FILE__, __LINE__, __func__, dbp, table_name.c_str());
		dbp->open(NULL,       // Txn pointer
                  config.dbfile.c_str(), // File name
                  table_name.c_str(),       // Logical db name
                  DB_BTREE, // Database type (using btree)
                  db_flags, // Open flags
                  0);         // File mode. Using defaults


	}
	catch(DbException& e) {
		this->freeDbp( dbp );

		char	msg[ 1024 ];
		snprintf(msg, sizeof(msg) - 1, "fail to open table '%s' (ENV:%s DBFILE:%s cache-size:%lu) -- %s", table_name.c_str(), config.envhome.c_str(), config.dbfile.c_str(), dbp_cache.size(), e.what() );
		throw std::runtime_error( msg );
	}

	// put dbp into cache
	dbp_type	dbp_removed = dbp_cache.put( table_name, dbp );
	if (dbp_removed) {
		dbp_removed->close( 0 );
		
//fprintf(stderr, "%s:%d %s() dbp_removed=%p\n", __FILE__, __LINE__, __func__, dbp_removed);
		this->freeDbp( dbp_removed );
	}

	return dbp;
}


void
SleepyStore::closeAllTables()
{
	boost::mutex::scoped_lock	_scoped_lock( this->dbp_cache_mutex );

	dbp_cache_type::Key_List	_list( dbp_cache.get_all_keys() );
	for( dbp_cache_type::Key_List_Iter liter = _list.begin(); liter != _list.end(); liter++ ) {
		dbp_type	dbp = dbp_cache.get( *liter, false );

		dbp->close( 0 );
		this->freeDbp( dbp );
	}

	dbp_cache.clear();
}

///////////////////////////////////////////////////////////////////////
//

SleepyStore::dbp_type
SleepyStore::getDbp()
{
	boost::mutex::scoped_lock	_scoped_lock( this->free_dbp_list_mutex );

	if (free_dbp_list.size() > 0) {
		dbp_type	dbp = free_dbp_list.front();
		free_dbp_list.pop_front();
		return dbp;
	}

	dbp_type	dbp = new Db( dbenv_p, 0 );
//fprintf(stderr, "%s:%d %s() dbp=%p\n", __FILE__, __LINE__, __func__, dbp);
	return dbp;
}

void
SleepyStore::freeDbp( dbp_type& dbp )
{
	boost::mutex::scoped_lock	_scoped_lock( this->free_dbp_list_mutex );

//fprintf(stderr, "%s:%d %s() dbp=%p\n", __FILE__, __LINE__, __func__, dbp);
	////free_dbp_list.push_back( dbp );
	delete dbp;
	dbp = NULL;
}

void
SleepyStore::purgeDbp()
{
	boost::mutex::scoped_lock	_scoped_lock( this->free_dbp_list_mutex );

	for (list_dbp_type::iterator it=free_dbp_list.begin(); 
			it != free_dbp_list.end();
			it++ )
	{
		dbp_type&	dbp = *it;
		delete dbp;
	}

	free_dbp_list.clear();
}



///////////////////////////////////////////////////////////////////////
//

SleepyConfig::SleepyConfig( const char* config_filename )
{
	this->parseConfig( config_filename );
}


int
SleepyConfig::parseConfig( const char* config_filename )
{
	/*
 	 *       INI config
	 *   [fecab::store]
	 *   envhome = /path/to/env/home
	 *   dbfile  = /path/to/db/file  [ default : sleepy.db ]
	 *   engine  = btree             [ default : btree ]
	 *   ntables = 65536             [ default : 65536 ]
	 *   btree.minkey = 64           [ default : 0 ]
	 *
	 *   ncache-tables = 1024  		 [ default : 1024 ]
	 */

	bocat::common::INIFile	ini( config_filename );
	
	std::string	envhome = ini[ INI_SECTION ][ "envhome" ];
	if (envhome.size() < 1) {
		char	msg[512];
		snprintf(msg, sizeof(msg) - 1, "No 'envhome' or [%s] in config file %s", INI_SECTION, config_filename);
		throw std::runtime_error( msg );
	}
	std::string	dbfile  = ini[ INI_SECTION ][ "dbfile" ];
	if (dbfile.size() < 1) {
		dbfile = DEFAULT_DB_FILENAME;
	}
	std::string	engine  = ini[ INI_SECTION ][ "engine" ];
	if (engine.size() < 1) {
		engine = DEFAULT_ENGINE;
	}

	this->envhome = envhome;
	this->dbfile = dbfile;
	this->engine = engine;

	std::string	ntables = ini[ INI_SECTION ][ "ntables" ];
	if (ntables.size() < 1) {
		this->ntables = DEFAULT_NTABLES;
	}
	else {
		this->ntables = atoi( ntables.c_str() );
	}

	std::string	btree_minkey = ini[ INI_SECTION ][ "btree.minkey" ];
	if (btree_minkey.size() < 1) {
		this->btree_minkey = 2;
	}
	else {
		this->btree_minkey = atoi( btree_minkey.c_str() );
	}

	std::string	ncache_tables = ini[ INI_SECTION ][ "ncache_tables" ];
	if (ncache_tables.size() < 1) {
		this->ncache_tables = DEFAULT_NCACHE_TABLES;
	}
	else {
		this->ncache_tables = atoi( ncache_tables.c_str() );
	}

    std::string recovery = ini[ INI_SECTION ][ "recovery" ];
    if (recovery.size() < 1) {
        this->in_recovery = false;
    }   
    else {
        //this->in_recovery = boost::lexical_cast< bool >( recovery.c_str() );
        char    ch = recovery[ 0 ];
        if ( ('1' == ch) || ('t' == ch) || ('T' == ch) || ('y' == ch) || ('Y' == ch) ) { 
            this->in_recovery = true;
        }   
        else {
            this->in_recovery = false;
        }   
    }   


	/*
	std::cout << " parse Config File " << config_filename << " :\n"
			  << "   envhome = " << config.envhome << "\n"
			  << "    dbfile = " << config.dbfile << "\n"
			  << "    engine = " << config.engine << "\n"
			  << "   ntables = " << config.ntables << "\n"
			  << "   btree.minkey = " << config.btree_minkey << "\n"
			  << "   ncache_tables = " << config.ncache_tables << "\n"
			  << "\n";
	*/

	return 0;
}

}	//// namespace fecab

